package com.stan.core.spark.userAction;

import com.alibaba.fastjson.JSONObject;
import com.stan.common.ConcatStrUtils;
import com.stan.common.DateUtils;
import com.stan.common.SparkUtils;
import com.stan.common.TaskUtils;
import com.stan.core.conf.ConfManager;
import com.stan.core.contants.Constants;
import com.stan.core.mapper.SessionDetailMapper;
import com.stan.core.mapper.TaskMapper;
import com.stan.core.mapper.Top10CategoryMapper;
import com.stan.core.mapper.Top10SessionMapper;
import com.stan.core.mapper.factory.MapperFactory;
import com.stan.core.spark.SparkCalculator;
import com.stan.core.vo.SessionDetail;
import com.stan.core.vo.Task;
import com.stan.core.vo.Top10Category;
import com.stan.core.vo.Top10Session;
import com.stan.simulate.SimulateArgs;
import com.stan.simulate.SimulateDataSource;
import org.apache.commons.lang.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.*;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.storage.StorageLevel;
import scala.Tuple2;
import com.google.common.base.Optional;
import java.io.Serializable;
import java.util.*;

/**
 * 用户行为分析功能
 */
public class UserActionAnalysisSpark implements SparkCalculator, Serializable {
    public static void main(String[] args) {
        UserActionAnalysisSpark userActionAnalysisSpark = new UserActionAnalysisSpark();
        userActionAnalysisSpark.run(args);
    }

    @Override
    public void run(String[] args) {
        System.out.println(SparkUtils.isLocal());

        MapperFactory mapperFactory = MapperFactory.getMapperFactory();
        TaskMapper taskMapper = mapperFactory.getMapper(TaskMapper.class);


        // TODO: 远程模式 + 优化
        SparkConf sparkConf = new SparkConf()
                .setMaster(ConfManager.getProperty(Constants.CONF_SPARK_MASTER))
                .setAppName(Constants.CONF_USER_VISIT_ACTION_APP_NAME+"-"+ DateUtils.getTodayDate());
        JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);

        // 创建Task
        Task task = new Task();
        task.setCreateTime(DateUtils.formatDate(new Date()));
        task.setTaskId(UUID.randomUUID().toString());
        System.out.println("当前任务ID："+task.getTaskId());
        task.setTaskType("用户行为数据分析");

        JSONObject taskParams = null;
        if(args.length <= 0 ){
            task.setTaskName(TaskUtils.genTaskName(true));
            taskParams = SimulateArgs.simulateTaskParamJSONObject(); // 模拟信息
            task.setTaskParams(taskParams.toJSONString());
        }else{
            task.setTaskName(TaskUtils.genTaskName(false));
            String strTaskParam = args[0];
            taskParams = JSONObject.parseObject(strTaskParam);
        }
        taskParams.put(Constants.PARAM_TASK_ID,task.getTaskId());
        task.setTaskParams(taskParams.toJSONString());
        task.setTaskStatus(Task.TASK_CREATED);
        taskMapper.insert(task);

        // 进行计算
        task.setStartTime(DateUtils.formatTime(new Date()));
        task.setTaskStatus(Task.TASK_STARTED);
        taskMapper.update(task);
        compute(javaSparkContext,task);

        // 计算完成，更新task信息
        task.setFinishTime(DateUtils.formatTime(new Date()));
        task.setTaskStatus(Task.TASK_FINISHED);
        taskMapper.update(task);

        System.out.println("当前任务Task ID为:"+task.getTaskId());
        javaSparkContext.close();
    }

    @Override
    public void compute(final JavaSparkContext javaSparkContext, final Task task) {

        JSONObject taskParams = JSONObject.parseObject(task.getTaskParams());

        SQLContext sqlContext = getSQLContext(javaSparkContext);
        boolean  local = SparkUtils.isLocal();
        // TODO: 测试 - 模拟生成数据
        if(local){
            SimulateDataSource.mock(javaSparkContext,sqlContext);
        }

        // 通过SQLContext将一定限制的行为数据导入RDD
        JavaRDD<Row> originUserActionRDD = SparkUtils.queryUserActionDataToRDDWithDateRange(sqlContext, taskParams);
        // TODO: 输出测试
//        originUserActionRDD.foreach(new VoidFunction<Row>() {
//            @Override
//            public void call(Row row) throws Exception {
//                System.out.println(row);
//            }
//        });
        System.out.println(originUserActionRDD.count());
        // 一对一进行映射
        JavaPairRDD<String,Row> session2RowRDD = originUserActionRDD.mapToPair(new PairFunction<Row, String, Row>() {
            @Override
            public Tuple2<String, Row> call(Row row) throws Exception {
                return new Tuple2<String, Row>(row.getString(0),row);
            }
        });

        // 按照session粒度进行聚合
        JavaPairRDD<String,String> sessionId2AggrInfo = aggreateBySession(session2RowRDD,sqlContext);
        sessionId2AggrInfo.foreachPartition(new VoidFunction<Iterator<Tuple2<String, String>>>() {
            @Override
            public void call(Iterator<Tuple2<String, String>> tuple2Iterator) throws Exception {
                List<SessionDetail> sessionDetails = new ArrayList<>();
                while(tuple2Iterator.hasNext()){
                    Tuple2<String,String> currentTuple = tuple2Iterator.next();
                    SessionDetail sessionDetail =new SessionDetail();
                    sessionDetail.setSessionId(currentTuple._1());
                    String aggrInfo = currentTuple._2;
                    Map<String,String> aggrInfoMap = ConcatStrUtils.concatStrToMap(aggrInfo,"|");
                    sessionDetail.setCity(aggrInfoMap.get(Constants.FIELD_CITY));
                    sessionDetail.setName(aggrInfoMap.get(Constants.FIELD_NAME));
                    sessionDetail.setAge(aggrInfoMap.get(Constants.FIELD_AGE));
                    sessionDetail.setSex(aggrInfoMap.get(Constants.FIELD_SEX));
                    sessionDetail.setUserId(aggrInfoMap.get(Constants.FIELD_USER_ID));
                    sessionDetail.setProfessional(aggrInfoMap.get(Constants.FIELD_PROFESSIONAL));
                    sessionDetail.setSearchKeywords(aggrInfoMap.get(Constants.FIELD_SEARCH_KEYWORDS));
                    sessionDetail.setClickCategoryIds(aggrInfoMap.get(Constants.FIELD_CLICK_CATEGORY_IDS));
                    sessionDetail.setClickProductIds(aggrInfoMap.get(Constants.FIELD_CLICK_PRODUCT_IDS));
                    sessionDetail.setStartTime(aggrInfoMap.get(Constants.FIELD_START_TIME));
                    sessionDetail.setEndTime(aggrInfoMap.get(Constants.FIELD_END_TIME));
                    sessionDetail.setUsername(aggrInfoMap.get(Constants.FIELD_USERNAME));
                    sessionDetail.setStepLength(aggrInfoMap.get(Constants.FIELD_STEP_LENGTH));
                    sessionDetail.setVisitLength(aggrInfoMap.get(Constants.FIELD_VISIT_LENGTH));
                    sessionDetails.add(sessionDetail);
                }

                SessionDetailMapper mapper = MapperFactory.getMapperFactory().getMapper(SessionDetailMapper.class);
                mapper.insertList(sessionDetails);
            }
        });


        // 过滤
        JavaPairRDD<String,String> filteredSession2AggrInfo = filterSession2AggrInfo(sessionId2AggrInfo,taskParams);
        // 通过taskParams条件过滤的 sessionId2Row
        JavaPairRDD<String,Row> filteredSession2row =  getFilteredSession2Detail(
                session2RowRDD,filteredSession2AggrInfo);
        filteredSession2row = filteredSession2row.persist(StorageLevel.MEMORY_ONLY());
//        filteredSession2row.foreach(new VoidFunction<Tuple2<String, Row>>() {
//            @Override
//            public void call(Tuple2<String, Row> stringRowTuple2) throws Exception {
//                System.out.println("row:"+stringRowTuple2);
//            }
//        });


        // 功能1 : 随机抽取session进行分析
        randomExtractAnalysis(filteredSession2AggrInfo,filteredSession2row);

        // 功能2 : 获得Top10热门品类
        // 输入数据:filteredSession2AggrInfo(session粒度，uid为key)
        // 格式:(95,clickProductIds=80,13,21,51,53,98|stepLength=14|searchKeywords=新辣道鱼火锅,国贸大厦,温泉|visitLength=-2982000|startTime=2019-04-27|sessionId=bda191f76f8e4281ad25596be5cd4e56|endTime=2019-04-27|clickCategoryIds=16,16,16,16,16,16,16,16,16,16,16,16,16,16)
        // 目标数据: 获得Top10的品类数据 ，包括点击量、下单量 和 支付数量
        // 思路:
        //  1.过滤：过滤出包含品类数据的数据
        //  2.统计：对品类的点击量、下单量、支付量进行统计，并对平类重复项进行过滤
        //  3.排序：对品类按照点击量、下单量、支付量进行二次排序，因此在二次排序之前，需要首先用Ordered的实现类封装一下，
        //      因为使用sortByKey,要求key对象所在类实现了scala.math.Ordered接口.
        //  4.提取：提取Top10的Key，并写入mysql,表的格式为 (taskid,categoryid,clickCount,orderCount,payCount)
        List<ComparableCategoryObject> top10Category = extractTop10Category(filteredSession2row);

        // 同步到数据库
        List<Top10Category> top10Categories = new ArrayList<Top10Category>();
        for (ComparableCategoryObject comparableCategoryObject : top10Category){
            Top10Category top10CategoryOne = new Top10Category();
            top10CategoryOne.setCategoryId(comparableCategoryObject.categoryId);
            top10CategoryOne.setTaskId(taskParams.getString(Constants.PARAM_TASK_ID));
            top10CategoryOne.setClickCount(comparableCategoryObject.clickCategoryCount);
            top10CategoryOne.setOrderCount(comparableCategoryObject.orderCategoryCount);
            top10CategoryOne.setDefrayCount(comparableCategoryObject.defrayCategoryCount);
            top10Categories.add(top10CategoryOne);
        }
        final MapperFactory mapperFactory = MapperFactory.getMapperFactory();
        if(top10Categories.size() > 0){
            Top10CategoryMapper top10CategoryMapper = mapperFactory.getMapper(Top10CategoryMapper.class);
            top10CategoryMapper.insertList(top10Categories);
        }

        // 功能3 : 获得top10活跃session
        // 输入数据:filteredSession2AggrInfo(session粒度，uid为key)
        // 格式:(95,clickProductIds=80,13,21,51,53,98|stepLength=14|searchKeywords=新辣道鱼火锅,国贸大厦,温泉|visitLength=-2982000|startTime=2019-04-27|sessionId=bda191f76f8e4281ad25596be5cd4e56|endTime=2019-04-27|clickCategoryIds=16,16,16,16,16,16,16,16,16,16,16,16,16,16)
        // 目标数据: 获得Top10活跃的session数据，包括最受欢迎的sessionid,clickCount
        // 思路: 获得访问Top10品类的点击量频次Top10的session(也就是100个session)
        //   1.提取Top10的品类ID，形成一个PairRDD,用于进行join
        //   2.计算Top10品类被各个session的点击次数:
        //          首先将session粒度数据中的各个categoryid的点击次数，格式为(categoryid,sessionid_count),自然需要flatMapToPair
        //          在与1.创建的PairRDD进行join,得到session粒度的Top10的categoryid的(categoryid,sessionid_count)
        //   3.按照(categoryid,sessionid_count)或得到每个
        //      首先进行聚合 (categoryid,iterator(sessionid_count,....)) 然后挑出每个categoryid对应的Top10点击量即可
        List<ComparableSessionObject> top10SessionByTop10Category =
                extractTop10Session(javaSparkContext,top10Category,filteredSession2row);

        List<Top10Session> top10Sessions = new ArrayList<Top10Session>();
        for(ComparableSessionObject comparableSessionObject : top10SessionByTop10Category){
            Top10Session top10SessionOne = new Top10Session();
            top10SessionOne.setSessionId(comparableSessionObject.getSessionId());
            top10SessionOne.setClickCount(comparableSessionObject.getTop10ClickCategoryCount());
            top10SessionOne.setTaskId(taskParams.getString(Constants.PARAM_TASK_ID));
            top10SessionOne.setCategoryId(comparableSessionObject.getCategoryId());
            top10Sessions.add(top10SessionOne);
        }
        Top10SessionMapper top10SessionMapper = mapperFactory.getMapper(Top10SessionMapper.class);
        if(top10Sessions.size() > 0){
            top10SessionMapper.insertList(top10Sessions);
        }

        // TODO: task - 完成时间
    }

    /**
     * 随机抽取分析
     * 思路:
     * 1.计算出每天每小时的session数量
     * 2.使用按时间比例随机抽取算法，计算出每天每小时要抽取session的索引
     * 3.遍历每天每小时的session，然后根据随机索引进行抽取
     */
    public static void randomExtractAnalysis(
            JavaPairRDD<String,String> filteredSession2AggrInfo,
            JavaPairRDD<String,Row> filteredSession2row){
        // 1.计算出每天每小时的session数量
        JavaPairRDD<String,Long> hour2SessionCountBySession = filteredSession2row.mapToPair(
                new PairFunction<Tuple2<String, Row>, String, Long>() {
                    @Override
                    public Tuple2<String, Long> call(Tuple2<String, Row> stringRowTuple2) throws Exception {
                        String actionTime = stringRowTuple2._2.getString(3);
                        String hour = DateUtils.getDateHour(actionTime);
                        return new Tuple2<String, Long>(hour,1L);
                    }
                }
        );

        Map<String,Object> hour2SessionCount = hour2SessionCountBySession.countByKey();

        // 2.将 <yaer-month-day hour,count> 调整为 <year-month-day,(hour,count)>
        Map<String,Map<String,Long>> dateHourCountMap = new HashMap<String,Map<String,Long>>();

        for(String dateAndHour : hour2SessionCount.keySet()){
            Long count = (Long)hour2SessionCount.get(dateAndHour);
            String date =  dateAndHour.split("_")[0];
            String hour = dateAndHour.split("_")[1];
            Map<String,Long> hourAndCount = dateHourCountMap.get(date);
            if(hourAndCount == null){
                hourAndCount = new HashMap<String,Long>();
                dateHourCountMap.put(date,hourAndCount);
            }
            hourAndCount.put(hour,count);
        }
        // 首先按照天数进行平分
        int extractNumPerDay = 100/dateHourCountMap.size();

        // 计算出这一天
        Map<String,Map<String,List<Integer>>> dateHourExtractMap = new HashMap<>();
        Random random = new Random();

        for(String date : dateHourCountMap.keySet()){
            Map<String,Long> hourCountMap = dateHourCountMap.get(date);

            // 计算出这一天的总session数量
            long sessionCount = 0L;
            for(long hourCount : hourCountMap.values()){
                sessionCount += hourCount;
            }

            Map<String,List<Integer>> hourExtractMap = dateHourExtractMap.get(date);
            if(hourExtractMap == null) {
                hourExtractMap = new HashMap<String, List<Integer>>();
                dateHourExtractMap.put(date, hourExtractMap);
            }

            // 遍历每个小时
            for(String hour : hourCountMap.keySet()){
                long count = hourCountMap.get(hour);

                //计算出每个小时的session数量，占据当前session比例，直接乘以要抽取的数量
                // 就可以计算出，当前小时需要抽取的session的数量
                int hourExtractNumber = (int)(((double)count/(double)sessionCount)*extractNumPerDay);
                if(hourExtractNumber > count) {
                    hourExtractNumber = (int)count;
                }

                // 先获取当前小时的存放的随机数的list
                List<Integer> extractIndexList = hourExtractMap.get(hour);
                if(extractIndexList == null){
                    extractIndexList = new ArrayList<>();
                    hourExtractMap.put(hour,extractIndexList);
                }

                // 生成随机数
                for(int i = 0; i < hourExtractNumber; i++){
                    int extractIndex = random.nextInt((int)count);
                    while (extractIndexList.contains(extractIndex)){
                        extractIndex = random.nextInt((int)count);
                    }
                    extractIndexList.add(extractIndex);
                }
            }
        }
    }


    /**
     * 获得top10 category 的每个 top10 session
     * @param top10Categorys
     * @param filteredSession2row
     * @return
     */
    public static List<ComparableSessionObject> extractTop10Session(
            JavaSparkContext jsc,
            List<ComparableCategoryObject> top10Categorys,
            JavaPairRDD<String,Row> filteredSession2row){
        // 1.组织top10CategoryPairRDD <categoryId,categoryId>
        List<Tuple2<String, String>> top10CategoryTuples = new ArrayList<>();
        for(ComparableCategoryObject comparableCategoryObject : top10Categorys){
            String categoryId = comparableCategoryObject.getCategoryId();
            top10CategoryTuples.add(new Tuple2<String, String>(categoryId,categoryId));
        }
        JavaPairRDD<String,String> top10CategoryPairRDD = jsc.parallelizePairs(top10CategoryTuples);
        top10CategoryPairRDD = top10CategoryPairRDD.distinct();

        // 2.计算top10品类被各个session的点击次数 (sessionId_categoryId,count)
        JavaPairRDD<String,String> clickCategoryId2SessionId = filteredSession2row.flatMapToPair(
                new PairFlatMapFunction<Tuple2<String, Row>, String, String>() {
                    @Override
                    public Iterable<Tuple2<String, String>> call(Tuple2<String, Row> stringRowTuple2) throws Exception {
                        List<Tuple2<String,String>> clickCategoryId2SessionIdList = new ArrayList<>();
                        Row row = stringRowTuple2._2();
                        String sessionId = row.getString(0);
                        String clickCategoryId = row.getString(6);
                        if(clickCategoryId != null && !"".equals(clickCategoryId)){
                            clickCategoryId2SessionIdList.add(new Tuple2<String, String>(clickCategoryId,sessionId));
                        }
                        return clickCategoryId2SessionIdList;
                    }
                }
        );
        // join
        JavaPairRDD<String,Long> sessionIdCategoryId2CountBySession = top10CategoryPairRDD.join(clickCategoryId2SessionId)
                .mapToPair(
                        new PairFunction<Tuple2<String, Tuple2<String, String>>, String, Long>() {
                            @Override
                            public Tuple2<String, Long> call(Tuple2<String, Tuple2<String, String>> stringTuple2Tuple2) throws Exception {
                                String clickCategoryId = stringTuple2Tuple2._1;
                                String sessionId = stringTuple2Tuple2._2._2;
                                return new Tuple2<String, Long>(sessionId+"_"+clickCategoryId,1L);
                            }
                        }
                );
        JavaPairRDD<String,Long> sessionIdCategoryId2Count = sessionIdCategoryId2CountBySession.reduceByKey(
                new Function2<Long, Long, Long>() {
                    @Override
                    public Long call(Long num1, Long num2) throws Exception {
                        return num1 + num2;
                    }
                }
        );



        // 3.TopN排序
        // <categoryId,(sessionId,count)> 只保留每个categoryId对应的Top10的(sessionId,count) [按照count排序]
        JavaPairRDD<String,Tuple2<String,Long>> sessionId2categoryIdCount = sessionIdCategoryId2Count.mapToPair(
                new PairFunction<Tuple2<String, Long>, String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Tuple2<String, Long>> call(Tuple2<String, Long> tuple2) throws Exception {
                        String sessionId = tuple2._1.split("_")[0];
                        String categoryId = tuple2._1.split("_")[1];
                        long count = tuple2._2;
                        return new Tuple2<String, Tuple2<String, Long>>(categoryId,new Tuple2<String, Long>(sessionId,count));
                    }
                }
        );

        JavaRDD<Iterable<ComparableSessionObject>> everyCategoryTop10Session = sessionId2categoryIdCount
                .groupByKey()
                .map(new Function<Tuple2<String, Iterable<Tuple2<String, Long>>>,Iterable<ComparableSessionObject>>() {
                    @Override
                    public Iterable<ComparableSessionObject> call(
                            Tuple2<String, Iterable<Tuple2<String, Long>>> tuple) throws Exception {
                        String categoryId = tuple._1;
                        PriorityQueue<ComparableSessionObject> top10Queue = new PriorityQueue<>();
                        Iterator<Tuple2<String,Long>> iterator = tuple._2.iterator();
                        while(iterator.hasNext()){
                            Tuple2<String,Long> sessionId2Count = iterator.next();
                            String sessionId = sessionId2Count._1;
                            long count = sessionId2Count._2;
                            ComparableSessionObject comparableSessionObject = new ComparableSessionObject();
                            comparableSessionObject.setCategoryId(categoryId);
                            comparableSessionObject.setSessionId(sessionId);
                            comparableSessionObject.setTop10ClickCategoryCount(count);
                            if(top10Queue.size() < 10){
                                top10Queue.add(comparableSessionObject);
                            }else{
                                ComparableSessionObject minCountSession = top10Queue.peek();
                                if(count > minCountSession.getTop10ClickCategoryCount()){
                                    top10Queue.poll();
                                    top10Queue.add(comparableSessionObject);
                                }
                            }
                        }
                        return top10Queue;
                    }
                });


        List<ComparableSessionObject> top10SessionByTop10Category = new ArrayList<>();

        for(Iterable<ComparableSessionObject> iterable : everyCategoryTop10Session.collect()){
            Iterator<ComparableSessionObject> iterator = iterable.iterator();
            while (iterator.hasNext()){
                top10SessionByTop10Category.add(iterator.next());
            }
        }


        return top10SessionByTop10Category;
    }


    /**
     * 因为filteredSession2AggrInfo中的session是经过taskParams所有参数筛选的
     * 因此需要它的key去连接原始数据sessionId2Row以保留符合taskParams条件的数据
     * @param sessionId2Row
     * @param filteredSession2AggrInfo
     */
    public static JavaPairRDD<String,Row> getFilteredSession2Detail(
            JavaPairRDD<String,Row> sessionId2Row,
            JavaPairRDD<String,String> filteredSession2AggrInfo){
        return sessionId2Row.join(filteredSession2AggrInfo).mapToPair(new PairFunction<Tuple2<String, Tuple2<Row, String>>, String, Row>() {
            @Override
            public Tuple2<String, Row> call(Tuple2<String, Tuple2<Row, String>> stringTuple2Tuple2) throws Exception {
                String sessionId = stringTuple2Tuple2._1;
                Row row = stringTuple2Tuple2._2._1;
                return new Tuple2<String, Row>(sessionId,row);
            }
        });
    }



    // 功能2 : 获得Top、10热门品类
    // 输入数据:filteredSession2AggrInfo(session粒度，uid为key)
    // 格式:(95,clickProductIds=80,13,21,51,53,98|stepLength=14|searchKeywords=新辣道鱼火锅,国贸大厦,温泉|visitLength=-2982000|startTime=2019-04-27|sessionId=bda191f76f8e4281ad25596be5cd4e56|endTime=2019-04-27|clickCategoryIds=16,16,16,16,16,16,16,16,16,16,16,16,16,16)
    // 目标数据: 获得Top10的品类数据 ，包括点击量、下单量 和 支付数量
    // 思路:
    //  1.过滤：过滤出包含品类数据的数据
    //  2.统计：对品类的点击量、下单量、支付量进行统计，并对平类重复项进行过滤
    //  3.排序：对品类按照点击量、下单量、支付量进行二次排序，因此在二次排序之前，需要首先用Ordered的实现类封装一下，
    //     因为使用sortByKey,要求key对象所在类实现了scala.math.Ordered接口.
    //  4.提取：提取Top10的Key，并写入mysql,表的格式为 (taskid,categoryid,clickCount,orderCount,payCount)
    public static List<ComparableCategoryObject> extractTop10Category(JavaPairRDD<String,Row> filteredSession2row){
        // 1.计算点击、下单、支付的品类的数量
        JavaPairRDD<String,Long> clickCategoryId2Count = getClickCategory2CountRDD(filteredSession2row);
        final JavaPairRDD<String,Long> orderCategoryId2Count = getOrderCategory2CountRDD(filteredSession2row);
        JavaPairRDD<String,Long> defrayCategoryId2Count = getDefrayCategory2CountRDD(filteredSession2row);

        // 2.计算所有点击、下单、支付过的categoryId,组织成Pair的形式便于join
        JavaPairRDD<String,String> categoryId2CategoryId = filteredSession2row.flatMapToPair(
                new PairFlatMapFunction<Tuple2<String, Row>, String, String>() {
                    @Override
                    public Iterable<Tuple2<String, String>> call(Tuple2<String, Row> stringRowTuple2) throws Exception {
                        List<Tuple2<String,String>> result = new ArrayList<>();

                        String clickCategoryId = stringRowTuple2._2.getString(6);
                        if(clickCategoryId != null && !"".equals(clickCategoryId)){
                            result.add(new Tuple2<String, String>(clickCategoryId,clickCategoryId));
                        }
                        String orderCategoryIds = stringRowTuple2._2.getString(8);
                        if(orderCategoryIds != null && !"".equals(orderCategoryIds)){
                            String[] orderCategoryIdsArr = orderCategoryIds.split(",");
                            for(String orderCategoryId : orderCategoryIdsArr){
                                result.add(new Tuple2<String, String>(orderCategoryId,orderCategoryId));
                            }
                        }
                        String defrayCategoryIds = stringRowTuple2._2.getString(10);
                        if(defrayCategoryIds != null && !"".equals(defrayCategoryIds)){
                            String[] defrayCategoryIdsArr = defrayCategoryIds.split(",");
                            for(String defrayCategoryId : defrayCategoryIdsArr){
                                result.add(new Tuple2<String, String>(defrayCategoryId,defrayCategoryId));
                            }
                        }
                        return result;
                    }
                }
        );
        // 去重
        categoryId2CategoryId = categoryId2CategoryId.distinct();

        // 3.合并数据
        // 将数据组织成 <categoryId,clickCount_orderCount_defrayCount>
        JavaPairRDD<String,String> categoryId2allCount = joinAllCategoryCount(categoryId2CategoryId,
                clickCategoryId2Count,orderCategoryId2Count,defrayCategoryId2Count);

        //  4.排序：对品类按照点击量、下单量、支付量进行二次排序，因此在二次排序之前，需要首先用Ordered的实现类封装一下，
        //     因为使用sortByKey,要求key对象所在类实现了scala.math.Ordered接口.
        // 到上面这一步，已经完成了 (clickCategoryId,count)的计算，但是我们是需要依据value进行排序
        // 思路是将这些信息重新封装在ComparableCategoryKey中,(ComparableCategoryKey)
        JavaRDD<ComparableCategoryObject> comparableCategoryObjectJavaRDD =
                categoryId2allCount.map(
                        new Function<Tuple2<String, String>, ComparableCategoryObject>() {
                            @Override
                            public ComparableCategoryObject call(Tuple2<String, String> stringStringTuple2) throws Exception {
                                String categoryId = stringStringTuple2._1;
                                String allCount = stringStringTuple2._2;
                                String[] tmpAllCountSplited = allCount.split("_");
                                Long clickCount = Long.valueOf(tmpAllCountSplited[0]);
                                Long orderCount = Long.valueOf(tmpAllCountSplited[1]);
                                Long defrayCount = Long.valueOf(tmpAllCountSplited[2]);
                                ComparableCategoryObject comparableCategoryObject =
                                        new ComparableCategoryObject();
                                comparableCategoryObject.setCategoryId(categoryId);
                                comparableCategoryObject.setClickCategoryCount(clickCount);
                                comparableCategoryObject.setOrderCategoryCount(orderCount);
                                comparableCategoryObject.setDefrayCategoryCount(defrayCount);
                                return comparableCategoryObject;
                            }
                        }
                );


        // 4.获取Top10热门品类并同步到数据库
//        List<Tuple2<ComparableCategoryKey,String>> top10Categorys =  comparableCategory2AllCountRDD.take(5);
        List<ComparableCategoryObject> top10Categorys = comparableCategoryObjectJavaRDD.top(10);
        // 返回
        return top10Categorys;
    }

    /** 3.合并数据
    * 将数据组织成 <categoryId,clickCount_orderCount_defrayCount>
    **/
    public static JavaPairRDD<String,String> joinAllCategoryCount(
            JavaPairRDD<String,String> categoryId2CategoryId,
            JavaPairRDD<String,Long> clickCategoryId2Count,
            JavaPairRDD<String,Long> orderCategoryId2Count,
            final JavaPairRDD<String,Long> defrayCategoryId2Count){

        JavaPairRDD<String,String> categoryId2clickCount_ = categoryId2CategoryId
                .leftOuterJoin(clickCategoryId2Count)
                .mapToPair(
                        new PairFunction<Tuple2<String, Tuple2<String,Optional<Long>>>, String, String>() {
                            @Override
                            public Tuple2<String, String> call(
                                    Tuple2<String, Tuple2<String,Optional<Long>>> stringTuple2Tuple2) throws Exception {
                                Tuple2<String, String> result;
                                String categoryId = stringTuple2Tuple2._1;
                                Optional<Long> optionalClickCount = stringTuple2Tuple2._2._2;
                                if(optionalClickCount.isPresent()){
                                    result = new Tuple2<String, String>(categoryId,optionalClickCount.get() + "_");
                                }else{
                                    result = new Tuple2<String, String>(categoryId,"0_");
                                }
                                return result;
                            }
                        }
                );

        JavaPairRDD<String,String> categoryId2ClickCount_orderCount_ = categoryId2clickCount_
                .leftOuterJoin(orderCategoryId2Count)
                .mapToPair(
                        new PairFunction<Tuple2<String, Tuple2<String, Optional<Long>>>, String, String>() {
                            @Override
                            public Tuple2<String, String> call(
                                    Tuple2<String, Tuple2<String, Optional<Long>>> stringTuple2Tuple2) throws Exception {
                                Tuple2<String,String> result;
                                String categoryId = stringTuple2Tuple2._1;
                                String clickCount_ = stringTuple2Tuple2._2._1;
                                Optional<Long> optionalOrderCount = stringTuple2Tuple2._2._2;
                                if(optionalOrderCount.isPresent()){
                                    result = new Tuple2<String, String>(
                                            categoryId,clickCount_+optionalOrderCount.get()+"_");
                                }else{
                                    result = new Tuple2<String, String>(
                                            categoryId,clickCount_ + "0_");
                                }
                                return result;
                            }
                        }
                );

        JavaPairRDD<String,String> categoryId2ClickCount_orderCount_defrayCount = categoryId2ClickCount_orderCount_
                .leftOuterJoin(defrayCategoryId2Count)
                .mapToPair(
                        new PairFunction<Tuple2<String, Tuple2<String, Optional<Long>>>, String, String>() {
                            @Override
                            public Tuple2<String, String> call(
                                    Tuple2<String, Tuple2<String, Optional<Long>>> stringTuple2Tuple2) throws Exception {
                                Tuple2<String,String> result ;
                                String categoryId = stringTuple2Tuple2._1;
                                String clickCount_orderCount_ = stringTuple2Tuple2._2._1;
                                Optional<Long> optionalDefrayCount = stringTuple2Tuple2._2._2;
                                if(optionalDefrayCount.isPresent()){
                                    result = new Tuple2<String, String>(
                                            categoryId,
                                            clickCount_orderCount_ + optionalDefrayCount.get()
                                    );
                                }else{
                                    result = new Tuple2<String, String>(
                                            categoryId,
                                            clickCount_orderCount_ + "0"
                                    );
                                }
                                return result;
                            }
                        }
                );
        return  categoryId2ClickCount_orderCount_defrayCount;
    }


    /**
     * 计算支付的品类及其数量
     * 其实其中涉及的计算与 getOrderCategory2CountRDD 完全一致
     * 但是与getClickCategory2CountRDD 不同，因为getClickCategory2CountRDD针对的是clickCategoryId
     * 而本方法与getOrderCategory2CountRDD正对的分别是clickCategoryIds与orderCategoryIds
     * @param filteredSession2row
     * @return
     */
    public static JavaPairRDD<String,Long> getDefrayCategory2CountRDD(
            JavaPairRDD<String,Row> filteredSession2row
    ){

        JavaPairRDD<String,Long> defrayCategoryId2CountBySession = filteredSession2row.flatMapToPair(
                new PairFlatMapFunction<Tuple2<String, Row>, String, Long>() {
                    @Override
                    public Iterable<Tuple2<String, Long>> call(Tuple2<String, Row> stringRowTuple2) throws Exception {
                        List<Tuple2<String,Long>> result = new ArrayList<>();
                        String defrayCategoryIds = stringRowTuple2._2.getString(10);
                        // 1.过滤掉支付品类为空的数据
                        if(defrayCategoryIds == null || "".equals(defrayCategoryIds)){
                            return new ArrayList<Tuple2<String,Long>>();
                        }

                        // 2.拆分进行统计: 因为defrayCategoryIds中使用`,`隔开有多个商品 例如: 11,23,541,31
                        String[] tmp = defrayCategoryIds.split(",");
                        for(String defrayCategoryId : tmp){
                            result.add(new Tuple2<String, Long>(defrayCategoryId,1L));
                        }
                        return result;
                    }
                }
        );

        JavaPairRDD<String,Long> defrayCategoryId2Count = defrayCategoryId2CountBySession.reduceByKey(
                new Function2<Long, Long, Long>() {
                    @Override
                    public Long call(Long num1, Long num2) throws Exception {
                        return num1 + num2;
                    }
                }
        );
        return defrayCategoryId2Count;
    }




    /**
     * 计算下单的品类及其数量
     * @param filteredSession2row
     * @return
     */
    public static JavaPairRDD<String,Long> getOrderCategory2CountRDD(
            JavaPairRDD<String,Row> filteredSession2row
    ){

        JavaPairRDD<String,Long> orderCategoryId2CountBySession = filteredSession2row.flatMapToPair(
                new PairFlatMapFunction<Tuple2<String, Row>, String, Long>() {
                    @Override
                    public Iterable<Tuple2<String, Long>> call(Tuple2<String, Row> stringRowTuple2) throws Exception {
                        List<Tuple2<String,Long>> result = new ArrayList<>();
                        String orderCategoryIds = stringRowTuple2._2.getString(8);
                        // 1.过滤掉下单品类为空的数据
                        if(orderCategoryIds == null || "".equals(orderCategoryIds)){
                            return new ArrayList<Tuple2<String,Long>>();
                        }

                        // 2.拆分进行统计: 因为orderCategoryIds中使用`,`隔开有多个商品 例如: 11,23,541,31
                        String[] tmp = orderCategoryIds.split(",");
                        for(String orderCategoryId : tmp){
                            result.add(new Tuple2<String, Long>(orderCategoryId,1L));
                        }
                        return result;
                    }
                }
        );

        JavaPairRDD<String,Long> orderCategoryId2Count = orderCategoryId2CountBySession.reduceByKey(
                new Function2<Long, Long, Long>() {
                    @Override
                    public Long call(Long num1, Long num2) throws Exception {
                        return num1 + num2;
                    }
                }
        );
        return orderCategoryId2Count;
    }


    /**
     * 获得点击品类及其数量
     * @param filteredSession2row
     * @return
     */
    public static JavaPairRDD<String,Long> getClickCategory2CountRDD(
            JavaPairRDD<String,Row> filteredSession2row
    ){
        // 1.过滤: 过滤出包含品类点击数据的数据
        JavaPairRDD<String,Row> filteredSession2rowContainsClickCategoryId =
                filteredSession2row.filter(new Function<Tuple2<String, Row>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, Row> stringRowTuple2) throws Exception {
                boolean result = true;
                String clickCategoryId = stringRowTuple2._2.getString(6);
                if(clickCategoryId == null || "".equals(clickCategoryId)){
                    result = false;
                }
                return true;
            }
        });

        //  2.统计：对品类的点击量进行统计，并对品类重复项进行过滤
        JavaPairRDD<String,Long> clickCategory2CountBySessionRDD = filteredSession2rowContainsClickCategoryId
                .mapToPair(new PairFunction<Tuple2<String, Row>, String, Long>() {
                    @Override
                    public Tuple2<String, Long> call(Tuple2<String, Row> stringRowTuple2) throws Exception {
                        String clickCategoryId = stringRowTuple2._2.getString(6);
                        return new Tuple2<String, Long>(clickCategoryId,1L);
                    }
                });

        JavaPairRDD<String,Long> clickCategory2Count = clickCategory2CountBySessionRDD.reduceByKey(
                new Function2<Long, Long, Long>() {
                    @Override
                    public Long call(Long num1, Long num2) throws Exception {
                        return num1 + num2;
                    }
                }
        );

        return clickCategory2Count;
    }



    /**
     * 过滤聚合信息
     * @param session2AggrInfo
     * @param taskParam
     * @return
     */
    public static JavaPairRDD<String,String> filterSession2AggrInfo(JavaPairRDD<String,String> session2AggrInfo,
                                                                            JSONObject taskParam){
        // 获得条件
        final Integer startAge = taskParam.getString(Constants.PARAM_FIELD_START_AGE) == null ?
                null : Integer.valueOf(taskParam.getString(Constants.PARAM_FIELD_START_AGE));
        final Integer endAge = taskParam.getString(Constants.PARAM_FIELD_END_AGE) == null ?
                null : Integer.valueOf(taskParam.getString(Constants.PARAM_FIELD_END_AGE));
        final String professional = taskParam.getString(Constants.PARAM_PROFESSIONALS);
        final String sex = taskParam.getString(Constants.PARAM_SEX);
        final String keywords = taskParam.getString(Constants.PARAM_KEYWORDS);
        final String categoryIds = taskParam.getString(Constants.PARAM_CATEGORY_IDS);

        // 进行条件的筛选
        JavaPairRDD<String,String> filteredSession2AggrInfo = session2AggrInfo.filter(new Function<Tuple2<String, String>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, String> stringStringTuple2) throws Exception {
                boolean result = true;
                Map<String,String> fieldsMap = ConcatStrUtils.concatStrToMap(stringStringTuple2._2,"|");
                Integer age = fieldsMap.get(Constants.FIELD_AGE) == null ?
                        null : Integer.valueOf(fieldsMap.get(Constants.FIELD_AGE));
                /**
                 * TODO: 检查包含关系，例如CategoryId是否只需要满足categoryIds的包含关系即可
                 */

                if(startAge != null && age < startAge){
                    result = false;
                }

                if(endAge != null && age > endAge){
                    result = false;
                }

                if(professional != null && !professional.equals(fieldsMap.get(Constants.FIELD_PROFESSIONAL))){
                    result = false;
                }

                if(sex != null && !sex.equals(fieldsMap.get(Constants.FIELD_SEX))){
                    result = false;
                }

                if(keywords != null && !keywords.equals(fieldsMap.get(Constants.FIELD_SEARCH_KEYWORDS))){
                    result = false;
                }

                if(categoryIds != null && !categoryIds.equals(fieldsMap.get(Constants.FIELD_CLICK_CATEGORY_IDS))){
                    result = false;
                }
                return result;
            }
        });

        return filteredSession2AggrInfo;
    }

    /**
     * 按照session粒度进行聚合
     * @param session2RowRDD
     * @return
     */
    public static JavaPairRDD<String,String> aggreateBySession(JavaPairRDD<String,Row> session2RowRDD,
                                                               SQLContext sqlContext) {
        JavaPairRDD<String,Iterable<Row>> session2RowsRDD = session2RowRDD.groupByKey();
        session2RowsRDD.foreach(new VoidFunction<Tuple2<String, Iterable<Row>>>() {
            @Override
            public void call(Tuple2<String, Iterable<Row>> stringIterableTuple2) throws Exception {
                System.out.println(stringIterableTuple2);
            }
        });


        JavaPairRDD<String,String> session2AggrInfo = session2RowsRDD.mapToPair(
                new PairFunction<Tuple2<String, Iterable<Row>>, String, String>() {
            @Override
            public Tuple2<String, String> call(Tuple2<String, Iterable<Row>> stringIterableTuple2) throws Exception {
                // 需要聚合的字段:  search_keyword , click_category_ids,click_product_id,
                // visit_length,step_length,date(start_time,end_time)
                String sessionId = stringIterableTuple2._1;
                StringBuffer searchKeywordsConcatStr = new StringBuffer();
                StringBuffer clickCategoryIdConcatStr = new StringBuffer();
                StringBuffer clickProductIdConcatStr = new StringBuffer();
                int visitLength = -1;
                int stepLength = 0;
                Date startTime = null;
                Date endTime = null;
                String userId = null;

                Iterator<Row> rowsIterator = stringIterableTuple2._2.iterator();
                while (rowsIterator.hasNext()){
                  Row row = rowsIterator.next();
                  String actionTime = row.getString(3);
                  String searchKeyword = row.getString(5);
                  String clickCategoryId = row.getString(6);
                  String clickProductId =row.getString(7);

                  if(StringUtils.isEmpty(userId)){
                      userId = row.getString(1);
                  }


                  if(startTime == null){
                    startTime = DateUtils.parseTime(actionTime);
                  }else{
                      Date date = DateUtils.parseTime(actionTime);
                      if(date.before(startTime)){
                          startTime = date;
                      }
                  }

                  if(endTime == null){
                      endTime = DateUtils.parseTime(actionTime);
                  }else{
                      Date date = DateUtils.parseTime(actionTime);
                      if(date.after(endTime)){
                          endTime = date;
                      }
                  }


                if(StringUtils.isNotEmpty(searchKeyword)){
                    searchKeywordsConcatStr.append(searchKeyword+",");
                }

                if(StringUtils.isNotEmpty(clickProductId)){
                    clickProductIdConcatStr.append(clickProductId + ",");
                }

                if(StringUtils.isNotEmpty(clickCategoryId)){
                    clickCategoryIdConcatStr.append(clickCategoryId + ",");
                }

                    stepLength++;

                }

                // 计算访问时长
                visitLength = (int)(endTime.getTime() - startTime.getTime());


                // 部分聚合统计结果
                sessionId = sessionId;
                String searchKeywords = ConcatStrUtils.trimSeqarator(searchKeywordsConcatStr.toString(),
                        ",");
                String clickCategoryIds = ConcatStrUtils.trimSeqarator(clickCategoryIdConcatStr.toString(),
                        ",");
                String clickProductIds = ConcatStrUtils.trimSeqarator(clickProductIdConcatStr.toString(),
                        ",");
                stepLength = stepLength;
                visitLength = visitLength;
                String startTimeStr = DateUtils.formatTime(startTime);
                String endTimeStr = DateUtils.formatTime(endTime);


                // 至此完成上述字段的聚合
                // 进行数据拼接
                Map<String,String> map = new HashMap<>();
                map.put(Constants.FIELD_SESSION_ID, sessionId);
                map.put(Constants.FIELD_SEARCH_KEYWORDS, searchKeywords);
                map.put(Constants.FIELD_CLICK_CATEGORY_IDS, clickCategoryIds);
                map.put(Constants.FIELD_CLICK_PRODUCT_IDS,clickProductIds);
                map.put(Constants.FIELD_STEP_LENGTH,stepLength+"");
                map.put(Constants.FIELD_VISIT_LENGTH,visitLength+"");
                map.put(Constants.FIELD_START_TIME,startTimeStr);
                map.put(Constants.FIELD_END_TIME,endTimeStr);
                String aggrInfo = ConcatStrUtils.mapToConcatStr(map,"|");
                return new Tuple2<String, String>(userId,aggrInfo);
            }
        });
        // 以上进行了session粒度的部分聚合，换用user_id的原因在: 需要使用user_id作为公共字段对user_info进行JOIN，然后把session_id
        // 换成key
        // 首先访问user_info的数据
        DataFrame dataFrame = sqlContext.sql("select * from user_info");
        JavaRDD<Row> userInfoRDD = dataFrame.javaRDD();

        JavaPairRDD<String,String> user2UserInfoRDD = userInfoRDD.mapToPair(new PairFunction<Row, String, String>() {
            @Override
            public Tuple2<String , String> call(Row row) throws Exception {
                String userId = row.getString(0);
                String username = row.getString(1);
                String name = row.getString(2);
                String age = row.getString(3);
                String professional = row.getString(4);
                String city = row.getString(5);
                String sex = row.getString(6);

                Map<String,String> map = new HashMap<>();
                map.put(Constants.FIELD_USERNAME,username);
                map.put(Constants.FIELD_NAME,name);
                map.put(Constants.FIELD_AGE,age);
                map.put(Constants.FIELD_PROFESSIONAL,professional);
                map.put(Constants.FIELD_CITY,city);
                map.put(Constants.FIELD_SEX,sex);
                String aggrInfo2 = ConcatStrUtils.mapToConcatStr(map,"|");
                return new Tuple2<String, String>(userId,aggrInfo2);
            }
        });

        JavaPairRDD<String,Tuple2<String,String>> userId2AggrInfoAB = session2AggrInfo.join(user2UserInfoRDD);
        JavaPairRDD<String,String> sessionId2AggrInfo = userId2AggrInfoAB.mapToPair(new PairFunction<Tuple2<String, Tuple2<String, String>>, String, String>() {
            @Override
            public Tuple2<String, String> call(Tuple2<String, Tuple2<String, String>> stringTuple2Tuple2) throws Exception {
                String userId = stringTuple2Tuple2._1;
                String sessionId = ConcatStrUtils.getFieldFromConcatStr(stringTuple2Tuple2._2._1,"|",Constants.FIELD_SESSION_ID);
                String aggrInfo = stringTuple2Tuple2._2._1+"|"+stringTuple2Tuple2._2._2;
                aggrInfo = ConcatStrUtils.addFieldFromConcatStr(aggrInfo,"|",Constants.FIELD_USER_ID,userId);
                return new Tuple2<String, String>(sessionId,aggrInfo);
            }
        });

        return sessionId2AggrInfo;
    }


    /**
     * 这里本地模式返回SQLContext对象,生产环境返回HiveContext对象
     * @param javaSparkContext
     * @return
     */
    public SQLContext getSQLContext(JavaSparkContext javaSparkContext){
        String sparkMaster = ConfManager.getProperty(Constants.CONF_SPARK_MASTER) ;
        // 本地模式
        if(SparkUtils.isLocal()){
            return new SQLContext(javaSparkContext.sc());
        }else { //生产环境运行
            return new HiveContext(javaSparkContext.sc());
        }
    }
}
